package com.fitz.remote.server;

import com.fitz.registry.InstanceInfo;
import com.fitz.registry.InstanceRegistry;
import com.fitz.remote.codec.RpcDecoder;
import com.fitz.remote.codec.RpcEncoder;
import com.fitz.remote.handler.RpcServerHandler;
import com.fitz.remote.model.BasicObject;
import com.fitz.remote.model.RpcRequest;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * @author fitz 2020-5-15 10:48
 */

public class ProviderServer{

    private static final Logger LOGGER = LoggerFactory.getLogger(ProviderServer.class);

    private ServerBootstrap serverBootstrap = new ServerBootstrap();

    private InstanceRegistry instanceRegistry;

    private Class<? extends BasicObject> clazz;

    private InstanceInfo instance;

    private Map<String, Object> beanMap;

    public ProviderServer(InstanceRegistry instanceRegistry, InstanceInfo instance, Map<String, Object> beanMap) {
        this.clazz = RpcRequest.class;
        this.instance = instance;
        this.instanceRegistry = instanceRegistry;
        this.beanMap = beanMap;
        init();
    }

    private void init(){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        Class<NioServerSocketChannel> channelClass = NioServerSocketChannel.class;

        serverBootstrap.group(bossGroup,workerGroup)
                .option(ChannelOption.SO_BACKLOG,120)
                .childOption(ChannelOption.TCP_NODELAY,true)
                .childOption(ChannelOption.SO_KEEPALIVE,true)
                .channel(channelClass)
                .childHandler(initializer());
    }

    private ChannelInitializer initializer(){
        return new ChannelInitializer() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                channel.pipeline()
                        .addLast("RpcDecoder",new RpcDecoder(instance.getSerializerType(),clazz))
                        .addLast("RpcEncoder",new RpcEncoder(instance.getSerializerType()))
                        .addLast(new RpcServerHandler(beanMap))
                ;
            }
        };
    }

    public void start(){
        try {
            serverBootstrap.bind(instance.getHost(),instance.getPort()).sync();
            LOGGER.info("Server bind [{}:{}]",instance.getHost(),instance.getPort());
            instanceRegistry.register(instance);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
